package com.leosam.tvbox.mv.proxy;

import java.io.File;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class M3u8DownloadExecutor extends ThreadPoolExecutor {

    private static class M3u8DownloadBlockingQueue extends AbstractQueue<Runnable> implements BlockingQueue<Runnable> {

        /**
         * 存储所有下载任务
         */
        private final LinkedHashMap<String, DownloadRunnable> tasks = new LinkedHashMap<>();

        /**
         * 存储需要优先下载的uri
         */
        private final Queue<String> priorityQueue = new LinkedList<>();

        /**
         * 存储需要下载的数据的启发式头,避免所有m3u8任务全部展开
         */
        private final Queue<String> headQueue = new LinkedList<>();

        /**
         * 对某个资源进行加急
         * 
         * @param uri 加急
         */
        private DownloadRunnable recordExpedited(String uri) {
            synchronized (this) {
                DownloadRunnable runnable = tasks.get(uri);
                if (runnable != null) {
                    priorityQueue.add(uri);
                }
                return runnable;
            }
        }

        /**
         * 标识某个资源的头
         * 
         * @param uri 加急
         */
        private DownloadRunnable recordHead(String uri) {
            synchronized (this) {
                DownloadRunnable runnable = tasks.get(uri);
                if (runnable != null) {
                    headQueue.add(uri);
                }
                return runnable;
            }
        }

        @Override
        public DownloadRunnable poll() {
            synchronized (this) {
                String pr;
                while ((pr = priorityQueue.poll()) != null) {
                    DownloadRunnable ret = tasks.remove(pr);
                    if (ret != null) {
                        // 优先下载下一个片段
                        if (tasks.containsKey(ret.getNextIndexName())) {
                            priorityQueue.add(ret.getNextIndexName());
                        }
                        return ret;
                    }
                }
                while ((pr = headQueue.poll()) != null) {
                    DownloadRunnable ret = tasks.remove(pr);
                    if (ret != null) {
                        // 优先下载下一个片段
                        if (tasks.containsKey(ret.getNextIndexName())) {
                            headQueue.add(ret.getNextIndexName());
                        }
                        return ret;
                    }
                }

                Iterator<DownloadRunnable> iterator = tasks.values().iterator();
                if (iterator.hasNext()) {
                    DownloadRunnable ret = iterator.next();
                    iterator.remove();
                    return ret;
                } else {
                    return null;
                }
            }
        }

        @Override
        public DownloadRunnable peek() {
            synchronized (this) {
                String pr;
                while ((pr = priorityQueue.peek()) != null) {
                    DownloadRunnable ret = tasks.get(pr);
                    if (ret != null) {
                        return ret;
                    }
                }
                while ((pr = headQueue.peek()) != null) {
                    DownloadRunnable ret = tasks.get(pr);
                    if (ret != null) {
                        return ret;
                    }
                }

                Iterator<DownloadRunnable> iterator = tasks.values().iterator();
                if (iterator.hasNext()) {
                    DownloadRunnable ret = iterator.next();
                    return ret;
                } else {
                    return null;
                }
            }
        }

        @Override
        public boolean offer(Runnable e) {
            if (!(e instanceof DownloadRunnable)) {
                return false;
            }
            synchronized (this) {
                tasks.put(e.toString(), (DownloadRunnable) e);
                this.notify();
            }
            return true;
        }

        @Override
        public void put(Runnable e) throws InterruptedException {
            offer(e);
        }

        @Override
        public boolean offer(Runnable e, long timeout, TimeUnit unit) throws InterruptedException {
            offer(e);
            return true;
        }

        @Override
        public DownloadRunnable take() throws InterruptedException {
            synchronized (this) {
                DownloadRunnable poll;
                while ((poll = poll()) == null) {
                    this.wait();
                }
                return poll;
            }
        }

        @Override
        public DownloadRunnable poll(long timeout, TimeUnit unit) throws InterruptedException {
            synchronized (this) {
                DownloadRunnable poll;
                while ((poll = poll()) == null) {
                    this.wait(unit.toMillis(timeout));
                }
                return poll;
            }
        }

        @Override
        public int remainingCapacity() {
            return Integer.MAX_VALUE;
        }

        @Override
        public int drainTo(Collection<? super Runnable> c) {
            synchronized (this) {
                int size = tasks.size();
                c.addAll(tasks.values());
                tasks.clear();
                priorityQueue.clear();
                return size;
            }
        }

        @Override
        public int drainTo(Collection<? super Runnable> c, int maxElements) {
            synchronized (this) {
                int i = 0;
                Iterator<DownloadRunnable> iterator = tasks.values().iterator();
                while (i < maxElements) {
                    if (iterator.hasNext()) {
                        c.add(iterator.next());
                        iterator.remove();
                        ++i;
                    } else {
                        break;
                    }
                }
                return i;
            }
        }

        @Override
        @SuppressWarnings("unchecked")
        public Iterator<Runnable> iterator() {
            return (Iterator<Runnable>) (Iterator<?>) tasks.values().iterator();
        }

        @Override
        public int size() {
            return tasks.size();
        }

    }

    /**
     * 运行中的任务.
     */
    private final ConcurrentHashMap<String, DownloadRunnable> running = new ConcurrentHashMap<>();

    public M3u8DownloadExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, new M3u8DownloadBlockingQueue());
        allowCoreThreadTimeOut(true);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        if (r instanceof DownloadRunnable) {
            DownloadRunnable runnable = (DownloadRunnable) r;
            running.put(runnable.getIndexName(), runnable);
        }
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        if (r instanceof DownloadRunnable) {
            DownloadRunnable runnable = (DownloadRunnable) r;
            running.remove(runnable.getIndexName(), runnable);
        }
    }

    /**
     * 对某个资源进行加急
     * 
     * @param uri 加急
     */
    public DownloadRunnable recordExpedited(String uri) {
        return ((M3u8DownloadBlockingQueue) getQueue()).recordExpedited(uri);
    }

    /**
     * 记录某个资源的头,在处理完优先下载的资源后,将下载这些资源
     * 
     * @param uri 加急
     */
    public DownloadRunnable recordHead(String uri) {
        return ((M3u8DownloadBlockingQueue) getQueue()).recordHead(uri);
    }

    /**
     * 阻塞当前线程直到文件下载完成
     * 
     * @param uri       下载的文件uri
     * @param storeBase 文件存储地址(基础路径,未拼接媒体名)
     * @return 如果下载成功返回file,下载失败返回null
     * @throws InterruptedException
     */
    public File waitForDownload(String uri, String storeBase, long timeout) throws InterruptedException {
        File file = new File(storeBase, uri);
        DownloadRunnable runnable = recordExpedited(uri);
        if (runnable == null) {
            runnable = running.get(uri);
            if (runnable == null) {
                if (file.exists()) {
                    return file;
                } else {
                    return null;
                }
            }
        }

        return waitForDownload(runnable, file, timeout);
    }

    public File waitForDownload(DownloadRunnable runnable, File file, long timeout) throws InterruptedException {
        synchronized (runnable) {
            if (file.exists()) {
                return file;
            }
            runnable.wait(timeout);
            if (file.exists()) {
                return file;
            } else {
                return null;
            }
        }
    }

    public File waitForDownload(DownloadRunnable runnable, long timeout) throws InterruptedException {
        File file = runnable.getFile();
        synchronized (runnable) {
            if (file.exists()) {
                return file;
            }
            runnable.wait(timeout);
            if (file.exists()) {
                return file;
            } else {
                return null;
            }
        }
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        throw new UnsupportedOperationException("M3u8DownloadExecutor unsupport submit()");
    }

    @Override
    public Future<?> submit(Runnable task) {
        throw new UnsupportedOperationException("M3u8DownloadExecutor unsupport submit()");
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        throw new UnsupportedOperationException("M3u8DownloadExecutor unsupport submit()");
    }
}
